{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Migrating from Spark to BigQuery via Dataproc -- Part 5\n",
    "\n",
    "* [Part 1](01_spark.ipynb): The original Spark code, now running on Dataproc (lift-and-shift).\n",
    "* [Part 2](02_gcs.ipynb): Replace HDFS by Google Cloud Storage. This enables job-specific-clusters. (cloud-native)\n",
    "* [Part 3](03_automate.ipynb): Automate everything, so that we can run in a job-specific cluster. (cloud-optimized)\n",
    "* [Part 4](04_bigquery.ipynb): Load CSV into BigQuery, use BigQuery. (modernize)\n",
    "* [Part 5](05_functions.ipynb): Using Cloud Functions, launch analysis every time there is a new file in the bucket. (serverless)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Catch-up cell"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "wget http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\n",
    "gunzip kddcup.data_10_percent.gz\n",
    "BUCKET='cloud-training-demos-ml'  # CHANGE\n",
    "gcloud storage cp kdd* gs://$BUCKET/\n",
    "bq mk sparktobq"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create reporting function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile main.py\n",
    "\n",
    "from google.cloud import bigquery\n",
    "import google.cloud.storage as gcs\n",
    "import tempfile\n",
    "import os\n",
    "\n",
    "def create_report(BUCKET, gcsfilename, tmpdir):\n",
    "    \"\"\"\n",
    "    Creates report in gs://BUCKET/ based on contents in gcsfilename (gs://bucket/some/dir/filename)\n",
    "    \"\"\"\n",
    "    # connect to BigQuery\n",
    "    client = bigquery.Client()\n",
    "    destination_table = 'sparktobq.kdd_cup'\n",
    "    \n",
    "    # Specify table schema. Autodetect is not a good idea for production code\n",
    "    job_config = bigquery.LoadJobConfig()\n",
    "    schema = [\n",
    "        bigquery.SchemaField(\"duration\", \"INT64\"),\n",
    "    ]\n",
    "    for name in ['protocol_type', 'service', 'flag']:\n",
    "        schema.append(bigquery.SchemaField(name, \"STRING\"))\n",
    "    for name in 'src_bytes,dst_bytes,wrong_fragment,urgent,hot,num_failed_logins'.split(','):\n",
    "        schema.append(bigquery.SchemaField(name, \"INT64\"))\n",
    "    schema.append(bigquery.SchemaField(\"unused_10\", \"STRING\"))\n",
    "    schema.append(bigquery.SchemaField(\"num_compromised\", \"INT64\"))\n",
    "    schema.append(bigquery.SchemaField(\"unused_12\", \"STRING\"))\n",
    "    for name in 'su_attempted,num_root,num_file_creations'.split(','):\n",
    "        schema.append(bigquery.SchemaField(name, \"INT64\")) \n",
    "    for fieldno in range(16, 41):\n",
    "        schema.append(bigquery.SchemaField(\"unused_{}\".format(fieldno), \"STRING\"))\n",
    "    schema.append(bigquery.SchemaField(\"label\", \"STRING\"))\n",
    "    job_config.schema = schema\n",
    "\n",
    "    # Load CSV data into BigQuery, replacing any rows that were there before\n",
    "    job_config.create_disposition = bigquery.CreateDisposition.CREATE_IF_NEEDED\n",
    "    job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE\n",
    "    job_config.skip_leading_rows = 0\n",
    "    job_config.source_format = bigquery.SourceFormat.CSV\n",
    "    load_job = client.load_table_from_uri(gcsfilename, destination_table, job_config=job_config)\n",
    "    print(\"Starting LOAD job {} for {}\".format(load_job.job_id, gcsfilename))\n",
    "    load_job.result()  # Waits for table load to complete.\n",
    "    print(\"Finished LOAD job {}\".format(load_job.job_id))\n",
    "    \n",
    "    # connections by protocol\n",
    "    sql = \"\"\"\n",
    "        SELECT COUNT(*) AS count\n",
    "        FROM sparktobq.kdd_cup\n",
    "        GROUP BY protocol_type\n",
    "        ORDER by count ASC    \n",
    "    \"\"\"\n",
    "    connections_by_protocol = client.query(sql).to_dataframe()\n",
    "    connections_by_protocol.to_csv(os.path.join(tmpdir,\"connections_by_protocol.csv\"))\n",
    "    print(\"Finished analyzing connections\")\n",
    "    \n",
    "    # attacks plot\n",
    "    sql = \"\"\"\n",
    "                            SELECT \n",
    "                             protocol_type, \n",
    "                             CASE label\n",
    "                               WHEN 'normal.' THEN 'no attack'\n",
    "                               ELSE 'attack'\n",
    "                             END AS state,\n",
    "                             COUNT(*) as total_freq,\n",
    "                             ROUND(AVG(src_bytes), 2) as mean_src_bytes,\n",
    "                             ROUND(AVG(dst_bytes), 2) as mean_dst_bytes,\n",
    "                             ROUND(AVG(duration), 2) as mean_duration,\n",
    "                             SUM(num_failed_logins) as total_failed_logins,\n",
    "                             SUM(num_compromised) as total_compromised,\n",
    "                             SUM(num_file_creations) as total_file_creations,\n",
    "                             SUM(su_attempted) as total_root_attempts,\n",
    "                             SUM(num_root) as total_root_acceses\n",
    "                           FROM sparktobq.kdd_cup\n",
    "                           GROUP BY protocol_type, state\n",
    "                           ORDER BY 3 DESC\n",
    "    \"\"\"\n",
    "    attack_stats = client.query(sql).to_dataframe()\n",
    "    ax = attack_stats.plot.bar(x='protocol_type', subplots=True, figsize=(10,25))\n",
    "    ax[0].get_figure().savefig(os.path.join(tmpdir,'report.png'));\n",
    "    print(\"Finished analyzing attacks\")\n",
    "    \n",
    "    bucket = gcs.Client().get_bucket(BUCKET)\n",
    "    for blob in bucket.list_blobs(prefix='sparktobq/'):\n",
    "        blob.delete()\n",
    "    for fname in ['report.png', 'connections_by_protocol.csv']:\n",
    "        bucket.blob('sparktobq/{}'.format(fname)).upload_from_filename(os.path.join(tmpdir,fname))\n",
    "    print(\"Uploaded report based on {} to {}\".format(gcsfilename, BUCKET))\n",
    "\n",
    "\n",
    "def bigquery_analysis_cf(data, context):\n",
    "    # check that trigger is for a file of interest\n",
    "    bucket = data['bucket']\n",
    "    name = data['name']\n",
    "    if ('kddcup' in name) and not ('gz' in name):\n",
    "        filename = 'gs://{}/{}'.format(bucket, data['name'])\n",
    "        print(bucket, filename)\n",
    "        with tempfile.TemporaryDirectory() as tmpdir:\n",
    "            create_report(bucket, filename, tmpdir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile requirements.txt\n",
    "google-cloud-bigquery\n",
    "google-cloud-storage\n",
    "pandas\n",
    "matplotlib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# verify that the code in the CF works\n",
    "name='kddcup.data_10_percent'\n",
    "if 'kddcup' in name and not ('gz' in name):\n",
    "    print(True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test that the function endpoint works"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# test that the function works\n",
    "import main as bq\n",
    "\n",
    "BUCKET='cloud-training-demos-ml' # CHANGE\n",
    "try:\n",
    "    bq.create_report(BUCKET, 'gs://{}/kddcup.data_10_percent'.format(BUCKET), \"/tmp\")\n",
    "except Exception as e:\n",
    "    print(e.errors)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy the cloud function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deploying function (may take a while - up to 2 minutes)...⠼"
     ]
    }
   ],
   "source": [
    "!gcloud functions deploy bigquery_analysis_cf --no-gen2 --region <region> --runtime python310 --trigger-resource $BUCKET --trigger-event google.storage.object.finalize"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Try it out\n",
    "\n",
    "Copy the file to the bucket:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud storage cp kddcup.data_10_percent gs://$BUCKET/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Verify that the Cloud Function is being run. You can do this from the [Cloud Functions](https://console.cloud.google.com/functions/) part of the GCP Console.\n",
    "\n",
    "Once the function is complete (in about 30 seconds), see if the output folder contains the report:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud storage ls gs://$BUCKET/sparktobq"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copyright 2019 Google Inc. Licensed under the Apache License, Version 2.0 (the \"License\"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
